package com.echo.im.storage.backup.mq;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.echo.im.common.vo.GroupMessageVO;
import com.echo.im.common.contant.MQConstants;

import com.echo.im.infrastructure.mq.event.BaseEvent;
import com.echo.im.repository.dao.entity.GroupMessage;
import com.echo.im.repository.dao.mapper.GroupMessageMapper;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * 私聊消息存储队列消费者 - 使用手动ACK和批量拉取、批量插入数据库
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class GroupStorageMessageReceiver {

    private final GroupMessageMapper groupMessageMapper;

    // 批量处理的最大消息数
    private static final int BATCH_SIZE = 200;

    /**
     * 监听RabbitMQ队列中的私聊消息
     * - 手动ACK模式由`ackMode = "MANUAL"`指定。
     * - 批量预取值在注解中的`prefetch`配置，设置为10，表示一次最多拉取10条消息。
     */
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = MQConstants.STORE_GROUP_QUEUE, durable = "true"),
                    exchange = @Exchange(value = MQConstants.ECHO_STORE_EXCHANGE, type = ExchangeTypes.TOPIC),
                    key = MQConstants.STORE_GROUP_TOPIC
            ),
            ackMode = "MANUAL",  // 手动ACK模式
            containerFactory = "rabbitListenerContainerFactoryWithPrefetch"  // 配置批量预取值
    )
    public void onChatMessage(List<Message> messages, Channel channel) throws Exception {

        List<Long> deliveryTags = new ArrayList<>();
        List<GroupMessage> groupMessagesBatch = new ArrayList<>();  // 用于批量插入的列表

        try {
            // 遍历处理从队列中拉取到的每条消息
            for (Message message : messages) {
                String msg = new String(message.getBody(), StandardCharsets.UTF_8);

                // 将消息体转换为自定义的业务对象
                BaseEvent.EventMessage<GroupMessageVO> eventMessage
                        = JSON.parseObject(msg, new TypeReference<BaseEvent.EventMessage<GroupMessageVO>>(){}.getType());

                GroupMessageVO data = eventMessage.getData();
                GroupMessage groupMessage = BeanUtil.copyProperties(data, GroupMessage.class);
                groupMessagesBatch.add(groupMessage);  // 添加到批量插入列表

                // 收集消息的deliveryTag以便批量确认
                deliveryTags.add(message.getMessageProperties().getDeliveryTag());

                // 如果已达到批量大小，立即进行数据库批量插入和批量ACK
                if (groupMessagesBatch.size() == BATCH_SIZE) {
                    insertAndAckBatch(groupMessagesBatch, deliveryTags, channel);
                }
            }

            // 处理完所有消息后，如果还有未插入和未ACK的消息，进行最后的批量操作
            if (!groupMessagesBatch.isEmpty()) {
                insertAndAckBatch(groupMessagesBatch, deliveryTags, channel);
            }

        } catch (Exception e) {
            e.printStackTrace();
            log.error("处理消息出现异常：{}", e.getMessage());
            // 如果出现异常，拒绝处理并重新投递这些消息
            for (Long deliveryTag : deliveryTags) {
                channel.basicNack(deliveryTag, false, false); // 重新投递
            }
        }
    }

    /**
     * 批量插入消息到数据库并确认已处理的消息
     *
     * @param groupMessagesBatch 本次批量处理的消息对象集合
     * @param deliveryTags 本次批量处理的消息的deliveryTag集合
     * @param channel 与RabbitMQ通信的信道
     * @throws Exception 批量操作时发生的异常
     */
    private void insertAndAckBatch(List<GroupMessage> groupMessagesBatch, List<Long> deliveryTags, Channel channel) throws Exception {
        try {
            // 批量插入数据库
            groupMessageMapper.insert(groupMessagesBatch,200);
            log.info("批量插入 {} 条消息到数据库", groupMessagesBatch.size());

            // 批量确认消息，使用最后一个deliveryTag进行批量确认
            long lastDeliveryTag = deliveryTags.get(deliveryTags.size() - 1);
            channel.basicAck(lastDeliveryTag, true);  // true表示确认所有小于等于此deliveryTag的消息
            log.info("批量ACK确认 {} 条消息", deliveryTags.size());

            // 清空已插入和确认的消息列表
            groupMessagesBatch.clear();
            deliveryTags.clear();
        } catch (Exception e) {
            log.error("批量插入或ACK出现异常：{}", e.getMessage());
            throw e;
        }
    }
}
